package NG::Sys::Spec::BlockerDaemon;
use strict;
use warnings;
use utf8;

use EV;
use JSON::XS;

use base 'NG::Sys::Loop::EVDaemon';
use NG::Sys::Sock qw(UnixListen InetListen CreateListener SendMessage);
use NG::Sys::Average;

my %all;
my $all = 0;
my %f2w;
my %cmd = (
		get => \&get_block,
		try => \&try_block,
		weak => \&try_block,
		free => \&free_block,
		stat => \&calc_stat,
	);

my $yes;
my $no;
my $timeout;
my %block;
my %timeout;
my $avg;
my $app;

#*check_block = sub{die "not defained check_block"};


sub on_prepare_loop {
	my ($self) = @_;
	unless($self->{blocker}){
		die "need config blocker";
	}
	my $sock;
	if($self->{blocker}{unix}){
		$sock = UnixListen($self->{cfg}{path}{var}."/".$self->{blocker}{unix}{path});
	}elsif($self->{blocker}{inet}){
		$sock = InetListen(@{$self->{blocker}{inet}}{'peer','port'});
	}else{
		die "need config blocker 'unix' or 'inet'";
	}
	if($self->{blocker}{type} eq "local"){
		*check_block = \&check_block_global;
		#*check_block = \&check_block_local;
	}elsif($self->{blocker}{type} eq "global"){
		*check_block = \&check_block_global;
	}else{
		die "invalid type block '$self->{blocker}{type}'";
	}
	$app = $self;

	$self->{connect} = CreateListener($sock,\&_on_connect,\&parser,\&_on_message,\&_on_close,\&_on_error);
	$yes = $self->{blocker}{yes};
	$no = $self->{blocker}{no}." ";
	$timeout = $self->{blocker}{timeout} || 5;
	$avg = new NG::Sys::Average(len => $self->{blocker}{avg_len}||100);

}

sub on_after_loop {
	my ($self) = @_;
	delete $self->{connect};
	%timeout = ();
	%block = ();
	$avg = undef;
	$app = undef;
	$self->SUPER::on_after_loop;
}
sub _on_connect {
	my $w = shift;
	$all{++$all} = $w;
	$w->data->[1] = $all;
}
sub parser {
	my $ref_str = shift;
	if((my $i = index($$ref_str,chr(0))) != -1){
		#my ($cmd,$key,$pid,$proc) = split / /,substr($w->data->[0],0,$i+1,""),4;
		my @res = split / /,substr($$ref_str,0,$i+1,""),4;
		return \@res;
	}
	return 0;
}
sub _on_message {
	my $w = shift;
	my ($cmd,$key,$pid,$proc) = @{shift()};
	if(my $sub = $cmd{$cmd}){
		$sub->($w,$key,$pid,$proc);
	}else{
		fail($w->fh,"bad cmd $cmd");
	}
}
sub ok {
	my $fh = shift;
	my $msg = shift || "";
	send_message($fh,$yes);
}
sub fail {
	my $fh = shift;
	my $msg = shift || "";
	send_message($fh,$no.$msg);
}
sub send_message {
	my $fh = shift;
	my $msg = shift;
	if(my $id = $f2w{$fh->fileno}){
		if(my $w = $all{$id}){
			$w->data->[0] .= $msg;
			return;
		}
	}
	my $w = SendMessage($fh,$msg,\&_on_close_send,\&_on_close_send);
	$all{++$all} = $w;
	$w->data->[1] = $all;
	$f2w{$fh->fileno} = $all;
}
sub _on_close {
	my $w = shift;
	delete $all{$w->data->[1]};
}
sub _on_close_send {
	my $w = shift;
	if(my $fn = $w->fh && $w->fh->fileno){
		delete $f2w{$fn};
	}
	delete $all{$w->data->[1]};
}
sub _on_error {
	my $w = shift;
	my $err = shift;
	$app->logger->error($err);
}
sub get_block {
	my ($w,$key,$pid,$proc) = @_;
	my $ary;
	$avg->count;
	unless($ary = $block{$key} and check_block($key)){
		$block{$key} = [ [$w->data->[1],time(),$pid,$proc] ];
		set_timeout($key);
		ok($w->fh);
	}else{
		push @$ary,[$w->data->[1],time(),$pid,$proc];
	}
}
sub weak_block {
	my ($w,$key,$pid,$proc) = @_;
	my $ary;
	$avg->count;
	unless($ary = $block{$key} and check_block($key)){
		$block{$key} = [ [$w->data->[1],time(),$pid,$proc] ];
		set_timeout($key);
		ok($w->fh);
	}else{
		my $f;
		for(@$ary){
			if($_->[3] eq $proc){
				$f = $_->[1].":".$_->[2];
				last;
			}
		}
		unless($f){
			push @$ary,[$w->data->[1],time(),$pid,$proc];
		}else{
			fail($w->fh,"buzy ".$f);
		}
	}
}
sub try_block {
	my ($w,$key,$pid,$proc) = @_;
	my $ary;
	$avg->count;
	unless($ary = $block{$key} and check_block($key)){
		set_timeout($key);
		$block{$key} = [ [$w->data->[1],time(),$pid,$proc] ];
		ok($w->fh);
	}else{
		fail($w->fh,"buzy ".join(":",@{$block{$key}[0]}[1,2]));
	}
}
sub free_block {
	my ($w,$key,$pid,$proc) = @_;
	my $ary;
	if($ary = $block{$key} and @$ary and $ary->[0][2] == $pid){
		shift @$ary;
		unless(check_block($key,1)){
			delete $timeout{$key};
			delete $block{$key};
		}
		ok($w->fh);
	}else{
		unless($ary && @$ary){
			delete $timeout{$key};
			delete $block{$key};
		}
		fail($w->fh,"not me");
	}
}
sub check_block_local {
	my $key = shift;
	my $ary = $block{$key};
	if(@_ == 0 and @$ary != 0){
		my $pid = $ary->[0][2];
		if(-d "/proc/$pid"){
			return 1;
		}
		shift @$ary;
	}
	while(@$ary){
		my $pid = $ary->[0][2];
		if(-d "/proc/$pid"){
			if(my $w = $all{$ary->[0][0]}){
				set_timeout($key);
				ok($w->fh);
				return 1;
			}
		}else{
			shift @$ary;
		}
	}
	return 0;
}
sub check_block_global {
	my $key = shift;
	my $ary = $block{$key};
	if(@_ == 0 and @$ary != 0){
		return 1;
	}
	while(@$ary){
		if(my $w = $all{$ary->[0][0]}){
			set_timeout($key);
			ok($w->fh);
			return 1;
		}else{
			shift @$ary;
		}
	}
	return 0;
}
sub set_timeout {
	my $key = shift;
	my $w;
	if($w = $timeout{$key}){
		$w->set($timeout);
	}else{
		$w = $timeout{$key} = EV::timer $timeout,0,\&call_timeout;
		$w->data($key);
	}
}
sub call_timeout {
	my $w = shift;
	my $key = $w->data;
	my $rec = shift @{$block{$key}};
	unless(check_block($key,1)){
		delete $timeout{$key};
		delete $block{$key};
	}
	$app->logger->warning("TIMEOUT: '".$key."' '".$rec->[1]."' '".$rec->[2]."' '".$rec->[3]);
}
sub calc_stat {
	my ($w,$key,$pid,$proc) = @_;
	my %res;
	my $c = 0;
	while(my($k,$v) = %block){
		if($res{$k} = @$v){
			$c++;
		}
	}
	my $str = encode_json({t=>\%res,c=>$c,s => $avg->stat});
	send_message($w->fh,$str);
}
1;
